package main

import (
	"encoding/json"
	"fmt"

	sarama "github.com/Shopify/sarama"
)

func UploadMessage(q chan TQuintet) {

	fmt.Println("producer启动")
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true
	config.Version = sarama.V0_11_0_2

	// producer, err := sarama.NewAsyncProducer([]string{KafkaIpAddress}, config)
	producer, err := sarama.NewAsyncProducer([]string{ClusterIp(Cluster) + KafkaPort}, config)
	if err != nil {
		fmt.Printf("producer_test create producer error :%s\n", err.Error())
		return
	}

	defer producer.AsyncClose()

	// send message
	msg := &sarama.ProducerMessage{
		Topic: "block",
		//Key:   sarama.StringEncoder("go_test"),
		Partition: 0,
	}
	for {
		b := <-q
		fmt.Println("上传五元组 ", b.Quin[0].BlockId)
		value, _ := json.Marshal(b)
		msg.Value = sarama.ByteEncoder(value)
		producer.Input() <- msg

		select {
		case suc := <-producer.Successes():
			fmt.Printf("offset: %d,  timestamp: %s \n", suc.Offset, suc.Timestamp.String())
		case fail := <-producer.Errors():
			fmt.Printf("err: %s\n", fail.Err.Error())
		}
	}

}
